package com.gzyj.flink.gps.map;

import com.gzyj.flink.dto.alarm.Kafka_VehicleNormalAlarm;
import com.gzyj.flink.gps.GPSPosition;
import com.gzyj.flink.gps.map.fatiguedriving.FatigueDrivingResult;
import com.gzyj.flink.gps.map.fatiguedriving.FatigueResolveDataProcess;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import java.util.Objects;

public class FatigueDrivingAlarmFlow {

    private SingleOutputStreamOperator<GPSPosition> gpsSingleStream;

    public FatigueDrivingAlarmFlow(SingleOutputStreamOperator<GPSPosition> gpsSingleStream) {
        this.gpsSingleStream = gpsSingleStream;
    }

    public void process() {

        SingleOutputStreamOperator<FatigueDrivingResult> fatigueDrivingResultSingleOutputStream = this.gpsSingleStream.map(new FatigueResolveDataProcess()).filter(Objects::nonNull);

        SingleOutputStreamOperator<Kafka_VehicleNormalAlarm> kafkaVehicleNormalAlarmSingleOutputStream = fatigueDrivingResultSingleOutputStream
                .filter(e -> e.getKafkaVehicleNormalAlarm() != null)
                .map(e -> e.getKafkaVehicleNormalAlarm());

        kafkaVehicleNormalAlarmSingleOutputStream.print();

        new BeidouAlarmProcess(kafkaVehicleNormalAlarmSingleOutputStream).process();

        fatigueDrivingResultSingleOutputStream
                .filter(e -> e.getKafkaVehicleRemindContent() != null)
                .map(e -> e.getKafkaVehicleRemindContent())
                .print();

    }


}
